package cn.com.bluemoon.bd.flink.func.map;

import cn.com.bluemoon.bd.flink.dim.processor.DimProcessorList;
import cn.com.bluemoon.bd.flink.dim.processor.IDimProcessor;
import cn.com.bluemoon.bd.flink.pojo.HourMinute;
import cn.com.bluemoon.bd.utils.DateUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;


import java.io.InputStream;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * 宽表生成抽象类
 *
 * 功能：
 * 1.关联所有相关的维表数据，生成宽表；
 * 2.在指定时间点加载更新维表数据。
 *
 * @param <IN> 流入数据的类型
 * @param <OUT> 输出数据的类型
 * @author 王仁线
 */
public abstract class AbstractWideMap<IN, OUT> extends RichMapFunction<IN, OUT> {
    private static final String DEFAULT_TIME_POINT = "23:30";
    private transient DimProcessorList<IN, OUT> dimProcessors;
    private transient ScheduledExecutorService executor;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.dimProcessors = new DimProcessorList<IN, OUT>();
        this.createDimProcessors();
        this.dimProcessors.open();

        /**
         * 启动维表加载的后台任务
         */
        this.executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                // 生成后台线程
                Thread t = Executors.defaultThreadFactory().newThread(r);
                t.setDaemon(true);
                return t;
            }
        });
        this.executor.scheduleAtFixedRate(new DimLoaderTask(this.dimProcessors, this.getTimePoints()), 10, 30, TimeUnit.SECONDS);
    }

    /**
     * 获取维表加载的时间点集合
     * @return
     */
    private String[] getTimePoints() {
        InputStream stream = null;
        String points = null;

        try {
            stream = Thread.currentThread().getContextClassLoader().getResourceAsStream("dim_load.properties");
            points = ParameterTool.fromPropertiesFile(stream).get("load.time.points");
            if (StringUtils.isBlank(points)) points = DEFAULT_TIME_POINT;
        } catch(Exception e) {
            points = DEFAULT_TIME_POINT;
        } finally {
            IOUtils.closeQuietly(stream);
        }

        String[] pointArr = points.split(",");
        return pointArr;
    }

    @Override
    public OUT map(IN input) throws Exception {
        OUT out = this.innerMap(input);
        this.dimProcessors.process(input, out);
        return out;
    }

    protected abstract OUT innerMap(IN input) throws Exception;

    protected abstract void createDimProcessors() throws Exception;

    public void addDimProcessor(IDimProcessor processor) {
        this.dimProcessors.addProcessor(processor);
    }

    @Override
    public void close() throws Exception {
        super.close();
        this.executor.shutdownNow();
        this.dimProcessors.close();
    }

    /**
     * 维表加载任务
     */
    private static class DimLoaderTask implements Runnable {
        // 加载时间点集合
        private Set<HourMinute> timePoints;
        // 维表处理器列表
        private DimProcessorList processors;

        private DimLoaderTask(DimProcessorList processors, String... hourMinutes) {
            this.processors = processors;
            this.timePoints = HourMinute.parse(hourMinutes);
        }

        @Override
        public void run() {
            try {
                String curHourMinute = DateUtils.dateFormat(new Date(), "HHmm");
                int hour = NumberUtils.toInt(curHourMinute.substring(0, 2));
                int minute = NumberUtils.toInt(curHourMinute.substring(2, 4));
                boolean exists = false;

                for (HourMinute hourMinute : this.timePoints) {
                    if (hourMinute.equalTimePoint(hour, minute)) {
                        exists = true;
                        break;
                    }
                }

                if (exists) {
                    this.processors.reload();
                    Thread.sleep(60 * 1000); // 加载完一次，休眠1分钟
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
